package com.bonc.hbase.api.ctrl;


import java.io.IOException;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Calendar;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import javax.servlet.http.HttpServletRequest;

import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.util.Bytes;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.util.StringUtils;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestMethod;
import org.springframework.web.bind.annotation.RestController;

import com.alibaba.fastjson.JSONArray;
import com.bonc.hbase.api.client.HbaseConvertUtil;
import com.bonc.hbase.api.client.HbaseOperateClient;
import com.bonc.hbase.api.constant.BaseConstant;
import com.bonc.hbase.api.entity.JsonResult;
import com.bonc.hbase.api.util.DateVerificationUtil;
import com.bonc.hbase.api.util.ValidateRequestUtil;


/**
 * 轨迹查询接口类
 * 
 * @author caiqiang
 * @version 2016年8月3日
 * @see TrackApiController
 * @since
 */
@RestController
@RequestMapping("/api/v1")
public class TrackApiController {
    /**
     * HBase 数据操作客户端
     */
    @Autowired
    private HbaseOperateClient hbaseOperateClient;

    /**
     * 日志对象
     */
    private static final Logger LOG = LoggerFactory.getLogger(TrackApiController.class);

    /**
     * Description:轨迹查询
     * 
     * @param phone
     * @param startDate
     * @param startHour
     * @param endDate
     * @param endHour
     * @param request
     * @return Object
     */
    @RequestMapping(value = {"/track"}, method = RequestMethod.GET)
    public Object getTrack(String phone, String startDate, String startHour, String endDate,
                           String endHour, HttpServletRequest request) {

        if (!ValidateRequestUtil.validate(request)) {
            return new JsonResult(BaseConstant.CODE_TOKEN_ILLEGAL, BaseConstant.MSG_TOKEN_ILLEGAL);
        }

        if (!ValidateRequestUtil.permissionValidate(phone)) {
            return new JsonResult(BaseConstant.CODE_INSUFFICIENT_PERMISSIONS,
                BaseConstant.MSG_INSUFFICIENT_PERMISSIONS);
        }

        if (isStrNullOrEmpty(phone, startDate, startHour, endDate, endHour)) {

            LOG.info("please input correct time ");
            return new JsonResult(BaseConstant.CODE_PARAM_INCOMPLETE,
                BaseConstant.MSG_PARAM_INCOMPLETE);
        }

        if (!DateVerificationUtil.verificationDate(startDate, endDate, startHour, endHour)) {
            return new JsonResult(BaseConstant.CODE_PARAM_ERROR, BaseConstant.MSG_PARAM_ERROR);
        }

        try {

        	byte[] start = convertToTrackRow(phone, startDate, startHour);
            byte[] end = convertToTrackRow(phone, endDate, endHour);
            Map<String, Integer> columnMap = new HashMap<String, Integer>();
            columnMap.put("longitude", 0);
            columnMap.put("latitude", 1);
            columnMap.put("createTime", 2);
            columnMap.put("provinceId", 3);
            columnMap.put("cityId", 4);
            
        	List<String> monthBetween = getMonthBetween(startDate,endDate);
        	JSONArray returnArrayResult = new JSONArray();
        	for(String monthStr:monthBetween){
        		String tableName = "track_"+monthStr;
                ResultScanner resultScanner = hbaseOperateClient.getResultScanner(start, end, tableName);
                JSONArray returnArray = HbaseConvertUtil.resultScan2JSONArray(columnMap,
                    resultScanner, "f", "q2");
                returnArrayResult.addAll(returnArray);
        	}
            return new JsonResult(returnArrayResult);
        }
        catch (IOException e) {
        	e.printStackTrace();
            LOG.info(e.getMessage());
            return new JsonResult(BaseConstant.CODE_ERROR, BaseConstant.MSG_ERROR);
        }
        catch (Exception e) {
        	e.printStackTrace();
            LOG.info(e.getMessage());
            return new JsonResult(BaseConstant.CODE_ERROR, BaseConstant.MSG_ERROR);
        }
    }

    /**
     * Description:创建rowkey
     * 
     * @param phone
     * @param date
     * @param hour
     * @return byte[]
     * @throws Exception
     * @see
     */
    private byte[] convertToTrackRow(String phone, String date, String hour)
        throws Exception {

        byte[] phoneId = Bytes.copy(Bytes.toBytes(Long.parseLong(phone)), 3, 5);
        byte[] deviceNumberHash = Bytes.toBytes((short)(phone.hashCode() & 0x7fff));
        byte[] phoneByte = Bytes.add(deviceNumberHash, phoneId);

        byte[] dateByte = Bytes.toBytes(Integer.parseInt(date));
        byte[] hourByte = Bytes.toBytes(Byte.parseByte(hour));

        byte[] rowkey = Bytes.add(phoneByte, dateByte);
        return Bytes.add(rowkey, hourByte);
    }

    /**
     * Description: 检查参数是否为空
     * 
     * @param phone
     * @param startDate
     * @param startHour
     * @param endDate
     * @param endHour
     * @return boolean
     * @see
     */
    private boolean isStrNullOrEmpty(String phone, String startDate, String startHour,
                                            String endDate, String endHour) {
        return StringUtils.isEmpty(phone) || StringUtils.isEmpty(startDate)
               || StringUtils.isEmpty(startHour) || StringUtils.isEmpty(endDate)
               || StringUtils.isEmpty(endHour);
    }
    
    private List<String> getMonthBetween(String startDate, String endDate) throws ParseException{
    	startDate = startDate.substring(0,6);
    	endDate = endDate.substring(0,6);
    	
		ArrayList<String> result = new ArrayList<String>();
		SimpleDateFormat sdf = new SimpleDateFormat("yyyyMM");//格式化为年月日

		Calendar min = Calendar.getInstance();
		Calendar max = Calendar.getInstance();

		min.setTime(sdf.parse(startDate));
		min.set(min.get(Calendar.YEAR), min.get(Calendar.MONTH), 1);

		max.setTime(sdf.parse(endDate));
		max.set(max.get(Calendar.YEAR), max.get(Calendar.MONTH), 2);

		Calendar curr = min;
		while (curr.before(max)) {
		 result.add(sdf.format(curr.getTime()));
		 curr.add(Calendar.MONTH, 1);
		}

		return result;
	}

}
